package com.example.flow

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import org.junit.Test
import kotlin.system.measureTimeMillis


/**
 * 流背压测试
 * 什么是背压？
 * 水流受到流动方向的压力。发射端发射速度大于收集端时收集速度，就会产生背压
 *
 *
 * @author zhouronghua
 * @time 2022/1/6 11:03 下午
 */
class FlowBackPressureTest {

    /**
     * 测试背压
     * 发射端每隔100毫秒发射一个元素，收集端每隔300毫秒收集一个元素
     *
     * @author zhouronghua
     * @time 2022/1/6 11:08 下午
     */
    @Test
    fun test_flow_back_pressure() = runBlocking {
        val myFlow = flow<Int> {
            for (i in 1..100) {
                delay(100)
                println("Emitting $i")
                emit(i)
            }
        }
        // 指定缓冲大小为100，缓冲池最大100个元素，超过100个等待收集再发送
        myFlow
            .buffer(100)
            .collect { value ->
            println("收集元素 $value")
            delay(300)
        }
    }

    /**
     * 测试背压
     * 发射端每隔100毫秒发射一个元素，收集端每隔300毫秒收集一个元素
     *
     * @author zhouronghua
     * @time 2022/1/6 11:08 下午
     */
    @Test
    fun test_flow_back_pressure_flowOn() = runBlocking {
        val myFlow = flow<Int> {
            for (i in 1..100) {
                delay(100)
                println("Emitting $i")
                emit(i)
            }
        }
        // 指定缓冲大小为100，缓冲池最大100个元素，超过100个等待收集再发送
        myFlow.flowOn(Dispatchers.Default)
            .buffer(100)
            .collect { value ->
                println("收集元素 $value")
                delay(300)
            }
    }

    /**
     * 测试背压conflate跳过中间的发射元素
     * 发射端每隔100毫秒发射一个元素，收集端每隔300毫秒收集一个元素
     * 当收集器处理它们太慢的时候， conflate 操作符可以用于跳过中间值
     *
     * @author zhouronghua
     * @time 2022/1/6 11:08 下午
     */
    @Test
    fun test_flow_back_pressure_conflate() = runBlocking {
        val myFlow = flow<Int> {
            for (i in 1..8) {
                delay(100)
                println("Emitting $i")
                emit(i)
            }
        }
        val time = measureTimeMillis {
            // 指定缓冲大小为100，缓冲池最大100个元素，超过100个等待收集再发送
            myFlow
//            .flowOn(Dispatchers.Default)
                .conflate()
                .collect { value ->
                    println("收集元素 $value ${Thread.currentThread().name}")
                    delay(300)
                }
        }
        println("Collected time $time")
    }

    /**
     * 测试背压collectLatest取消缓慢的收集器，并在每次发射新值的时候重新启动它。
     * 发射端每隔100毫秒发射一个元素，收集端每隔300毫秒收集一个元素
     * 当收集器处理它们太慢的时候， conflate 操作符可以用于跳过中间值
     *
     * @author zhouronghua
     * @time 2022/1/6 11:08 下午
     */
    @Test
    fun test_flow_back_pressure_collectLatest() = runBlocking {
        val myFlow = flow<Int> {
            for (i in 1..8) {
                delay(100)
                println("Emitting $i")
                emit(i)
            }
        }
        val time = measureTimeMillis {
            // 指定缓冲大小为100，缓冲池最大100个元素，超过100个等待收集再发送
            myFlow
                .collectLatest { value ->
                    println("收集元素 $value ${Thread.currentThread().name}")
                    delay(300)
                }
        }
        println("Collected time $time")
    }
}